package com.atguigu.flink0922.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/10 14:25
 */
public class Flink05_SQL_Kafka_Flink_Mysql {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        final StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        
        // 使用sql api 从kafka读取数据
        tenv.executeSql("create table sensor_1(" +
                            "   id string, " +
                            "   ts bigint, " +
                            "   vc int " +
                            ")with(" +
                            "   'connector' = 'kafka', " +
                            "   'topic' = 'sensor_1', " +
                            "   'properties.bootstrap.servers' = 'hadoop162:9092', " +
                            "   'properties.group.id' = 'Flink05_SQL_Kafka_Flink_Kafka', " +
                            "   'scan.startup.mode' = 'latest-offset', " +
                            "   'format' = 'csv'" +
                            ")");
        
        /*tenv.executeSql("create table sensor_2(" +
                            "   id string primary key, " +
                            "   sum_vc int " +
                            ")with(" +
                            "   'connector' = 'jdbc', " +
                            "   'url' = 'jdbc:mysql://hadoop162:3306/test?useSSL=false', " +
                            "   'table-name' = 'my_sensor', " +
                            "   'username' = 'root', " +
                            "   'password' = 'aaaaaa' " +
                            ")");*/
        tenv.executeSql("create table sensor_2(" +
                            "   id string, " +
                            "   sum_vc int," +
                            "   PRIMARY  KEY (id) NOT ENFORCED" +
                            ")with(" +
                            "   'connector' = 'jdbc', " +
                            "   'url' = 'jdbc:mysql://hadoop162:3306/test?useSSL=false', " +
                            "   'table-name' = 'my_sensor', " +
                            "   'username' = 'root', " +
                            "   'password' = 'aaaaaa' " +
                            ")");
        tenv.sqlQuery("select id, sum(vc) sum_vc from sensor_1 group by id").executeInsert("sensor_2");
        
    }
}
